package ex.other;


import ex.datastream.functions.function.FlatMapFuncBySplitter02;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


public class NacCatMysqlUndo {
    private StreamExecutionEnvironment env;
    private StreamTableEnvironment tableEnv;

    private void getEnv() {
        this.env = StreamExecutionEnvironment.getExecutionEnvironment();
        this.tableEnv = StreamTableEnvironment.create(env);
    }

    public static void main(String[] args) throws Exception {
        NacCatMysqlUndo demo = new NacCatMysqlUndo();
        demo.getEnv();
        DataStreamSource<String> dataStream = demo.env
                .socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> items = dataStream.flatMap(new FlatMapFuncBySplitter02());
        KeyedStream<Tuple2<String, Integer>, Object> keyBy = items.keyBy(value -> value.f0);
        WindowedStream<Tuple2<String, Integer>, Object, TimeWindow> window = keyBy.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));
        DataStream<Tuple2<String, Integer>> out = window.sum(1);
        System.out.println("out" + out.equals(null));
        Table inputTable = demo.tableEnv.fromDataStream(out);
        System.out.println("inputTable" + inputTable.equals(null));
        demo.tableEnv.createTemporaryView("result2", inputTable);

        Table resultTable2 = demo.tableEnv.sqlQuery("select * from result2");
        DataStream<Row> resultStream = demo.tableEnv.toDataStream(resultTable2);

        resultStream.printToErr();
        demo.registerMysql("test1", "test");
        demo.registerConsole();
        demo.env.execute("Window WordCount");
    }


    private Table registerMysql(String tablename, String flinkTableName) {
        String createOutDDL = "CREATE TABLE " + flinkTableName + " (" +
                "a STRING, " +
                "b BIGINT " +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=UTC&useSSL=false'," +
                "   'table-name' = '" + tablename + "'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' " +
                ")";
        tableEnv.executeSql(createOutDDL);
        Table ft = tableEnv.from(flinkTableName);
        return ft;
    }


    private void registerConsole() {
        String createPrintOutDDL = "CREATE TABLE printOutTable (" +
                "a STRING, " +
                "b BIGINT " +
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";
        tableEnv.executeSql(createPrintOutDDL);
    }

}
